package com.shujia.tour

import java.text.SimpleDateFormat
import java.util.Date

import org.apache.spark.sql.{DataFrame, SaveMode, SparkSession}

object CityDayIndex {
  def main(args: Array[String]): Unit = {

    /**
      * 市游客按天宽表
      *
      */

    /**
      *
      * d_city_id string comment '旅游目的地市代码'
      * ,d_city_name string comment '旅游目的地市名'
      * ,o_city_id string comment '旅游来源地地市代码'
      * ,o_city_name string comment '旅游来源地地市名'
      * ,o_province_id string comment '旅游来源地地省代码'
      * ,o_province_name string comment '旅游来源地省名'
      * ,number_attr string comment '号码归属地'
      * ,d_distance_section string comment '出游距离'
      * ,d_stay_time int comment '停留时间按小时'
      * ,gender string comment '性别'
      * ,trmnl_brand string comment '终端品牌'
      * ,pckg_price int comment '套餐'
      * ,conpot int comment '消费潜力'
      * ,age int comment '年龄'
      * ,only_pt int comment '电信游客量'
      *
      */


    if (args.length == 0) {
      println("请指定时间参数")
      return
    }

    var day_id: String = args(0)

    if ("%y%m%d".equals(day_id)) {
      val f: SimpleDateFormat = new SimpleDateFormat("yyyyMMdd")
      day_id = f.format(new Date)
    }
    println("当前分区时间：" + day_id)


    val month_id: String = day_id.substring(0, 6)
    println("当前月分区：" + month_id)

    val spark: SparkSession = SparkSession
      .builder()
      .appName("MergeLocationJob")
      .enableHiveSupport() //使用hived的元数据
      .getOrCreate()

    import spark.implicits._
    import org.apache.spark.sql.functions._

    /**
      * 出游距离分段
      * 10-20
      * 20-30
      * 30-40
      * 40-50
      * 50-60
      * 60-
      */


    val resultDF: DataFrame = spark.sql(
      s"""
         |
         |select * ,count(*) as only_pt
         |from
         |(select
         |e.d_city_id ,
         |e.d_city_name,
         |e.o_city_id,
         |e.o_city_name,
         |e.o_province_id,
         |e.o_province_name,
         |f.number_attr,
         |case
         |when e.d_max_distance < 20 then '20-'
         |when e.d_max_distance < 30 then '20-30'
         |when e.d_max_distance < 40 then '30-40'
         |when e.d_max_distance < 50 then '40-50'
         |when e.d_max_distance < 60 then '50-60'
         |else '60+' end as d_distance_section,
         |round(e.d_stay_time) as d_stay_time,
         |case
         |when f.gender='1' then '男'
         |when f.gender='2' then '女'
         |else '其他' end as gender,
         |f.trmnl_brand,
         |f.packg as pckg_price,
         |f.conpot,
         |f.age
         |from
         |(select c.*,
         |d.city_id as o_city_id,
         |d.city_name as o_city_name ,
         |d.prov_id as o_province_id ,
         |d.prov_name as o_province_name
         |from
         |(select a.*,b.city_name as d_city_name from
         |(select * from dal_tour.dal_tour_city_tourist_msk_d where day_id=$day_id) as a
         |join
         |(select distinct city_id,city_name from dim.dim_admincode) as b
         |on a.d_city_id=b.city_id) as c
         |join dim.dim_admincode as d
         |on c.source_county_id=d.county_id ) as e
         |join
         |(select * from dim.dim_usertag_msk_m where month_id=$month_id) as f
         |on e.mdn=f.mdn
         |) as g
         |group by
         |d_city_id
         |,d_city_name
         |,o_city_id
         |,o_city_name
         |,o_province_id
         |,o_province_name
         |,number_attr
         |,d_distance_section
         |,d_stay_time
         |,gender
         |,trmnl_brand
         |,pckg_price
         |,conpot
         |,age
         |
         |
      """.stripMargin)


    resultDF.write
      .format("csv")
      .option("sep", "\t")
      .mode(SaveMode.Overwrite)
      .save(s"/daas/motl/dal_tour/dal_tour_city_day_index/day_id=$day_id")


  }
}
